package com.yeming.flink.practice.state

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.streaming.api.scala._

/**
 * 第一种方法实现
 * 统计每个手机的呼叫时间间隔，时间单位是毫秒
 */
object TestKeyedState2 {

  def main(args: Array[String]): Unit = {
    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    //读取数据源
    val fileFile: String = getClass.getResource("/station.log").getPath
    val stream: DataStream[String] = streamEnv.readTextFile(fileFile)
    //数据计算，计算通话时间，开始时间和结束时间
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    streamLog.keyBy(_.callOut)
      //状态有两种情况，一种是有上一次的通话时间，另外一种就是没有上一次的通话时间。采用Scala的模式匹配
      .mapWithState[(String,Long), StationLog]{
        case (in:StationLog,None) => ((in.callOut,0L),Some(in))
        case (in:StationLog,pre: Some[StationLog]) => ((in.callOut,in.callTime - pre.get.callTime),Some(in))
      }
      .filter(_._2 != 0)
      .print("与上一次通话间隔：")

    //执行流计算
    streamEnv.execute()
  }

}
